package me.lijf.mqlite.config;

import lombok.extern.log4j.Log4j2;
import me.lijf.mqlite.redis.stream.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.util.Collections;

@Log4j2
@Configuration
public class RedisStreamConfig {
    @Autowired
    private StreamQueListener listener;
    @Autowired
    private StreamQueListener2 listener2;
    @Autowired
    private StreamQueListener3 listener3;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private StreamQueConfig config;

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options() {
        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .build();
    }

    //消息分发路由(cosumer-->listener)
    @Bean
    StreamMessageListenerContainer<String, MapRecord<String, String, String>> container(RedisConnectionFactory factory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options) {
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer.create(factory, options);
        for (RedisMqStream stream : config.getStreams()) {
            if (!redisTemplate.hasKey(stream.getName())) {
                redisTemplate.opsForStream().add(stream.getName(), Collections.singletonMap("", ""));
                log.info("Stream {} created.", stream.getName());
            }
            for (RedisMqGroup group : stream.getGroups()) {
                try {
                    redisTemplate.opsForStream().createGroup(stream.getName(), group.getName());
                } catch (Exception e) {
                    log.info("消费者组{}已存在", group.getName());
                }
            }
                /*
                for (String consumer : group.getConsumers()) {
                    log.info("绑定{}和{}",consumer,listener.getClass().getName());
                    container.receive(
                            Consumer.from(group.getName(), consumer),
                            StreamOffset.create(stream.getName(), ReadOffset.lastConsumed()),
                            listener
                    );
                }
                */
            }
        log.info("绑定{}和{}","lijf3_1_1",listener.getClass().getName());
        container.receive(
                Consumer.from("lijf3_1", "lijf3_1_1"),
                StreamOffset.create("lijf3", ReadOffset.lastConsumed()),
                listener
        );
        log.info("绑定{}和{}","lijf3_2_1",listener2.getClass().getName());
        container.receive(
                Consumer.from("lijf3_2", "lijf3_2_1"),
                StreamOffset.create("lijf3", ReadOffset.lastConsumed()),
                listener2
        );
        log.info("绑定{}和{}","lijf3_2_2",listener3.getClass().getName());
        container.receive(
                Consumer.from("lijf3_2", "lijf3_2_2"),
                StreamOffset.create("lijf3", ReadOffset.lastConsumed()),
                listener3
        );
        container.start();

        return container;
    }
}
